package com.gsm.partition;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO DataStream实现Wordcount：读socket（无界流）
 *
 * @author gsm
 * @version 1.0
 */
public class PartitionDemo {

    public static void main(String[] args) throws Exception {

        // TODO 1. 创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(2);

//        socket这种并行度永远为1 虽然设置了全局并行度是2 但是对socket不起作用
        DataStreamSource<String> socketDS =
                env.socketTextStream("localhost", 7777);

        // shuffle随机分区: random.nextInt(下游算子并行度) 跟源码！
//        socketDS.shuffle().print();

        // rebalance轮询：nextChannelToSendTo = (nextChannelToSendTo + 1) % 下游算子并行度
        // 如果是 数据源倾斜场景（比如kafka3个分区 下游3个并行度 正常理解 1-1 但是假设kafka某个分区数据量特别大 就会数据倾斜）
        // source后，调用rebalance，就可以解决 数据源的 数据倾斜
//        socketDS.rebalance().print();

        //rescale缩放： 实现轮询， 局部组队，比rebalance更高效
//        上游2 -> 下游4 如果是rebalance的话 上游1-> 1 2 3 4 上游2-> 1 2 3 4
//        上游2 -> 下游4 如果是rescale的话 上游1-> 1 2  上游2-> 3 4
//        socketDS.rescale().print();


        // broadcast 广播：  发送给下游所有的子任务 通常来说一条数据只给某个下游 这个不是 博爱
//        socketDS.broadcast().print();

        // global 全局： 全部发往 第一个子任务 通过webUI可以清晰查看到
        // return 0; 源代码写死了 0 这个分区
        socketDS.global().print();

        // keyby: 按指定key去发送，相同key发往同一个子任务
        // one-to-one: Forward分区器


        // 总结： Flink提供了 7种分区器+ 1种自定义

        env.execute();
    }
}

/**
    测试

         C:\Users\11706>nc -lp 7777
         1
         2
         3
         4
         5
         6
         7

        IDEA控制台输出(shuffle)
         2> 1
         1> 2
         2> 3
         1> 4
         2> 5
         1> 6
         1> 7   shuffle 效果

 */